package com.hucais.sync.es2hive.service

import com.hucais.core.constant.Constants
import com.hucais.core.utils.{CommonUtils, DefaultPropertiesUtil}
import com.hucais.etl.common.service.CommonQueryService
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.elasticsearch.spark.sql.EsSparkSQL

object SnycOdsDataService {
  private val esQuery =
    """
      |{
      |  "query": {
      |    "match_all": {}
      |  }
      |}
       """.stripMargin

  private val bookHdfsPath: String = "/hucaisdata/published/ods/book_information/"
  private val storeHdfsPath: String = "/hucaisdata/published/ods/book_store/"
  private val commentHdfsPath: String = "/hucaisdata/published/ods/book_comment/"

  def action(sparkSession: SparkSession): Unit = {
    import sparkSession.implicits._
   val bookEsDS = EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.information"), esQuery).as[OdsBook]
    val storeEsDS = EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.store"), esQuery).as[OdsStore]
    val commentEsDS = EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.comment"), esQuery).as[OdsComment]

    val bookDS = bookEsDS.mapPartitions(partitions => {
      partitions.map(item => {
        val isbn = CommonUtils.handleBlankStr(item.isbn)
        val book_name = CommonUtils.handleBlankStr(item.book_name)
        val author = CommonUtils.handleBlankStr(item.author)
        val pricing = CommonUtils.handleBlankStr(item.pricing)
        val category = CommonUtils.handleBlankStr(item.category)
        val publishing_house = CommonUtils.handleBlankStr(item.publishing_house)
        val publishing_time = CommonUtils.handleBlankStr(item.publishing_time)
        val impression = CommonUtils.handleBlankStr(item.impression)
        val edition = CommonUtils.handleBlankStr(item.edition)
        val sales_volume = CommonUtils.handleBlankStr(item.sales_volume)
        val binding_layout = CommonUtils.handleBlankStr(item.binding_layout)
        val format = CommonUtils.handleBlankStr(item.format)
        val number_of_pages = CommonUtils.handleBlankStr(item.number_of_pages)
        val paper = CommonUtils.handleBlankStr(item.paper)
        val number_of_suit = CommonUtils.handleBlankStr(item.number_of_suit)
        val data_source = CommonUtils.handleBlankStr(item.data_source)
        val acquisition_time = CommonUtils.handleBlankStr(item.acquisition_time)
        val acquisition_timestamp = CommonUtils.handleBlankStr(item.acquisition_timestamp)

        OdsBook(
          isbn, book_name, author, pricing, category, publishing_house,
          publishing_time, impression, edition, sales_volume, binding_layout,
          format, number_of_pages, paper, number_of_suit, data_source,
          acquisition_time, acquisition_timestamp
        )
      })
    })

    val storeDS = storeEsDS.mapPartitions(partitions => {
      partitions.map(item => {
        val isbn = CommonUtils.handleBlankStr(item.isbn)
        val name_of_shop = CommonUtils.handleBlankStr(item.name_of_shop)
        val slogan = CommonUtils.handleBlankStr(item.slogan)
        val book_review = CommonUtils.handleBlankStr(item.book_review)
        val selling_price = CommonUtils.handleBlankStr(item.selling_price)
        val store_pricing = CommonUtils.handleBlankStr(item.store_pricing)
        val conditions = CommonUtils.handleBlankStr(item.conditions)
        val number = CommonUtils.handleBlankStr(item.number)
        val data_source = CommonUtils.handleBlankStr(item.data_source)
        val acquisition_time = CommonUtils.handleBlankStr(item.acquisition_time)
        val acquisition_timestamp = CommonUtils.handleBlankStr(item.acquisition_timestamp)

        OdsStore(
          isbn, name_of_shop, slogan, book_review, selling_price,
          store_pricing, conditions, number, data_source, acquisition_time,
          acquisition_timestamp
        )
      })
    })

    val commentDS = commentEsDS.mapPartitions(partitions => {
      partitions.map(item => {
        val store_name = CommonUtils.handleBlankStr(item.store_name)
        val book_name = CommonUtils.handleBlankStr(item.book_name)
        val username = CommonUtils.handleBlankStr(item.username)
        val comment_content = CommonUtils.handleBlankStr(item.comment_content)
        val comment_time = CommonUtils.handleBlankStr(item.comment_time)
        val score = CommonUtils.handleBlankStr(item.score)
        val data_source = CommonUtils.handleBlankStr(item.data_source)
        val acquisition_time = CommonUtils.handleBlankStr(item.acquisition_time)
        val acquisition_timestamp = CommonUtils.handleBlankStr(item.acquisition_timestamp)

        OdsComment(
          store_name, book_name, username, comment_content, comment_time,
          score, data_source, acquisition_time, acquisition_timestamp
        )
      })
    })

    CommonQueryService.saveAsFileAbsPath(bookDS.toDF(), bookHdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
    CommonQueryService.saveAsFileAbsPath(storeDS.toDF(), storeHdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
    CommonQueryService.saveAsFileAbsPath(commentDS.toDF(), commentHdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
  }

  case class OdsBook(
                      isbn: String,
                      book_name: String,
                      author: String,
                      pricing: String,
                      category: String,
                      publishing_house: String,
                      publishing_time: String,
                      impression: String,
                      edition: String,
                      sales_volume: String,
                      binding_layout: String,
                      format: String,
                      number_of_pages: String,
                      paper: String,
                      number_of_suit: String,
                      data_source: String,
                      acquisition_time: String,
                      acquisition_timestamp: String
                    )

  case class OdsStore(
                       isbn: String,
                       name_of_shop: String,
                       slogan: String,
                       book_review: String,
                       selling_price: String,
                       store_pricing: String,
                       conditions: String,
                       number: String,
                       data_source: String,
                       acquisition_time: String,
                       acquisition_timestamp: String
                     )

  case class OdsComment(
                         store_name: String,
                         book_name: String,
                         username: String,
                         comment_content: String,
                         comment_time: String,
                         score: String,
                         data_source: String,
                         acquisition_time: String,
                         acquisition_timestamp: String
                       )

}
